package com.cti.commons.mqtt.config;

import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.stream.Collectors;

import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.env.Environment;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

import com.cti.commons.mqtt.annotation.Topic;
import com.cti.commons.mqtt.domain.PahoTopic;
import com.cti.commons.mqtt.domain.SubscriptTopic;
import com.cti.commons.mqtt.handle.MqttCallback;

import cn.hutool.core.util.StrUtil;
import lombok.Data;

@Data
@Configuration
@ConfigurationProperties(prefix = "commons.paho.mqtt")
public class MqttPahoConfiguration {

    public static final int MAX_IN_FLIGHT = 100;

    private String serverUri;

    private String username;

    private char[] password;

    private int maxInflight = MAX_IN_FLIGHT;

    private String clientId;

    private int completionTimeout;

    private int qos;

    private String subClientId;

    private Map<String, PahoTopic> topics;

    @Autowired
    private Environment environment;

    public MqttPahoConfiguration() {
    }

    @Bean
    public MqttConnectOptions getOption() {
        MqttConnectOptions options = new MqttConnectOptions();
        options.setServerURIs(new String[]{serverUri});
        options.setUserName(username);
        options.setPassword(password);
        options.setCleanSession(true);
        options.setAutomaticReconnect(true);
        options.setMaxInflight(maxInflight);
        return options;
    }

    @Bean
    public MqttPahoClientFactory mqttClientFactory(MqttConnectOptions options) {
        final DefaultMqttPahoClientFactory factory = new DefaultMqttPahoClientFactory();
        factory.setConnectionOptions(options);
        return factory;
    }

    @Bean
    public IntegrationFlow mqttOutFlow(MqttConnectOptions options) {
        return f -> f.handle(mqttOutbound(options));
    }

    /**
     * mqtt消息出站通道默认配置，用于向外发出mqtt消息
     *
     * @param options
     * @return
     */
    @Bean
    @ServiceActivator(inputChannel = "mqttOutputChannel")
    public MessageHandler mqttOutbound(MqttConnectOptions options) {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(clientId + "-" + UUID.randomUUID(),
            mqttClientFactory(options));
        messageHandler.setAsync(true);
        messageHandler.setDefaultQos(1);
        return messageHandler;
    }

    /**
     * mqtt消息出站通道，用于发送出站消息
     *
     * @return
     */
    @Bean
    public MessageChannel mqttOutputChannel() {
        return new DirectChannel();
    }

    @Bean
    public MqttAsyncClient getClient(MqttConnectOptions options,
        ApplicationContext applicationContext) throws Exception {
        MqttAsyncClient client = new MqttAsyncClient(serverUri, clientId + "-" + UUID.randomUUID(),
            new MemoryPersistence());
        List<SubscriptTopic> subscriptTopics = buildSubscriptTopic(applicationContext);
        client.setCallback(new MqttCallback(subscriptTopics));
        client.connect(options);
        return client;
    }

    private List<SubscriptTopic> buildSubscriptTopic(ApplicationContext applicationContext) {
        Map<String, Object> beansWithAnnotation = applicationContext.getBeansWithAnnotation(Topic.class);
        return beansWithAnnotation.keySet().stream().map(it -> {
            Class<?> classByteCode = beansWithAnnotation.get(it).getClass();
            Topic topic = AnnotationUtils.findAnnotation(classByteCode, Topic.class);
            PahoTopic pahoTopic = new PahoTopic(environment.resolvePlaceholders(topic.value()),
                topic.qos(), topic.mode(), topic.group());
            if (StrUtil.isEmpty(pahoTopic.getTopic())) {
                return null;
            }
            return new SubscriptTopic(pahoTopic, (IMqttMessageListener) applicationContext
                .getBean(classByteCode));
        }).filter(it -> Objects.nonNull(it)).collect(Collectors.toList());
    }

}
